package com.navinfo.opentsp.platform.computing.serializer;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Charsets;
import com.navinfo.opentsp.platform.computing.dto.*;
import com.navinfo.opentsp.platform.computing.util.DateUtils;
import com.navinfo.opentsp.platform.location.protocol.common.LCPredictiveMaintenance;
import com.navinfo.platform.bigdata.dto.DbcData;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.serialization.EventSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.msgpack.MessagePack;
import java.io.IOException;
import java.io.OutputStream;

/**
* DBCUpload信息
* HDFS_Sink的DBCUpload序列化器，拼接成字段间通过\001分割的字符串
*/
public class HDFSDBCUploadSerializer implements EventSerializer {
    private static final Logger logger = LoggerFactory.getLogger(HDFSDBCUploadSerializer.class);

    private final String APPEND_NEWLINE = "appendNewline";
    private final boolean appendNewline;
    private final String splitStr = "\001";
    private final OutputStream out;

    private MessagePack msgpack = new MessagePack();
    private HDFSDBCUploadSerializer(OutputStream out, Context ctx) {
        this.appendNewline = ctx.getBoolean(APPEND_NEWLINE, true);
        this.out = out;
    }

    @Override
    public boolean supportsReopen() {
        return false;
    }
    @Override
    public void afterCreate() {}
    @Override
    public void afterReopen() {}
    @Override
    public void beforeClose() {}
    @Override
    public void flush() throws IOException {
        this.out.flush();
    }
    @Override
    public void write(Event event) throws IOException {
        try {
//            logger.error(event.toString());
            //拼写落盘到HDFS的字符串。字段间分隔符是\001，null通过\N表示
            DbcData dbcdata = msgpack.read(event.getBody(), DbcData.class);
            if(dbcdata.getStaticsDataMap().isEmpty()){
                dbcdata.setStaticsDataMap(null);
            }
            DBCUploadDto dbcUploadDto = new DBCUploadDto();
            dbcUploadDto.setTerminalId(dbcdata.getTerminalId());
            dbcUploadDto.setSerialNumber(dbcdata.getSerialNumber());
            dbcUploadDto.setRetransmission(dbcdata.getRetransmission());
            dbcUploadDto.setVersion(dbcdata.getVersion());
            dbcUploadDto.setDepotLogo(dbcdata.getDepotLogo());
            dbcUploadDto.setReceiveDate(dbcdata.getReceiveDate());
            dbcUploadDto.setGpsDate(dbcdata.getGpsTime());
            dbcUploadDto.setLatitude(dbcdata.getLatitude());
            dbcUploadDto.setLongitude(dbcdata.getLongitude());
            dbcUploadDto.setOriginalLng(dbcdata.getOriginalLng());
            dbcUploadDto.setOriginalLat(dbcdata.getOriginalLat());
            dbcUploadDto.setDirection(dbcdata.getDirection());
            dbcUploadDto.setHeight(dbcdata.getHeight());
            dbcUploadDto.setStaticsDataMap(dbcdata.getStaticsDataMap());

            int part_time = Integer.parseInt(DateUtils.format(dbcdata.getGpsTime() * 1000, "yyyyMMdd"));
            dbcUploadDto.setPart_time(part_time);
            dbcUploadDto.setHashtid((int)(dbcdata.getTerminalId() % 32));
             String dataJson =JSON.toJSONString(dbcUploadDto);
            //添加字段只能在分割线之上添加,注意时间分区这块，终端传的时候是秒还是毫秒
//            logger.error(dataJson);
            //将报文解析后的内容拼接后写入hdfs
            this.out.write(dataJson.getBytes(Charsets.UTF_8));
            if (this.appendNewline) {
                this.out.write(10);
            }

        }catch (Exception e){
            logger.error("执行HDFS的HDFSPredictiveMaintenanceSerializer拦截器失败：" + e.getMessage(), e);
        }
    }

//    public static List<String> getMaintenanceEvents(List<LCLocationData.StaytimeParkingAddition> staytimeParkings){
//        List<String> staytimeParkingDtos = Lists.newArrayList();
//        if(staytimeParkings == null){
//            return staytimeParkingDtos;
//        }
//        for(LCLocationData.StaytimeParkingAddition staytimeParking : staytimeParkings){
//            staytimeParkingDtos.add(StaytimeParkingAdditionDto.parseString(staytimeParking));
//        }
//        return staytimeParkingDtos;
//    }


    public static class Builder implements EventSerializer.Builder {
        public Builder() { }
        @Override
        public EventSerializer build(Context context, OutputStream out) {
            return  new HDFSDBCUploadSerializer(out, context);
        }
    }
}
